package com.example.grpc.proxy;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http2.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class GrpcClientHandler extends ChannelInboundHandlerAdapter {

    private final Http2HeadersFrame http2HeadersFrame;
    private final Http2DataFrame dataFrame;
    private final ChannelHandlerContext outboundContext;

    private final Http2GoAwayFrame goAwayFrame = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR);

    public GrpcClientHandler(Http2HeadersFrame http2HeadersFrame, Http2DataFrame dataFrame, ChannelHandlerContext outboundContext) {
        this.http2HeadersFrame = http2HeadersFrame;
        this.dataFrame = dataFrame;
        this.outboundContext = outboundContext;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof Http2StreamFrame) {
            Http2StreamFrame streamFrame = (Http2StreamFrame) msg;
            log.info("stream id:{}, state: {}", streamFrame.stream().id(), streamFrame.stream().state());
        }
        // 判断stream是否endStream
        if (msg instanceof Http2DataFrame) {
            Http2DataFrame dataFrame = (Http2DataFrame) msg;
            int readableBytes = dataFrame.content().readableBytes();
            log.info("接收到 Http2DataFrame:{}", dataFrame.content().getCharSequence(0, readableBytes, CharsetUtil.UTF_8));
            outboundContext.writeAndFlush(dataFrame);
            if (dataFrame.isEndStream()) {
                outboundContext.writeAndFlush(goAwayFrame);
                ctx.writeAndFlush(goAwayFrame);
                log.info("服务端响应结束");
            }
        } else if (msg instanceof Http2HeadersFrame) {
            log.info("接收到 Http2HeadersFrame");
            Http2HeadersFrame headersFrame = (Http2HeadersFrame) msg;
            outboundContext.writeAndFlush(headersFrame);
            if (headersFrame.isEndStream()) {
                outboundContext.writeAndFlush(goAwayFrame);
                ctx.writeAndFlush(goAwayFrame);
                log.info("服务端响应结束");
            }

        } else if (msg instanceof Http2SettingsFrame) {
            Http2SettingsFrame settingsFrame = (Http2SettingsFrame) msg;
            log.info("接收到 Http2SettingsFrame:{}", settingsFrame.name());
        } else if (msg instanceof Http2SettingsAckFrame) {
            Http2SettingsAckFrame ackFrame = (Http2SettingsAckFrame) msg;
            log.info("接收到 Http2SettingsAckFrame:{}", ackFrame.name());
            ctx.write(http2HeadersFrame);
            ctx.writeAndFlush(dataFrame);
        } else if (msg instanceof Http2GoAwayFrame) {
            log.info("服务端主动关闭了连接.");
        } else if (msg instanceof Http2PingFrame){
            Http2PingFrame pingFrame = (Http2PingFrame) msg;
            log.info("接收到 PING FRAME received:{}", pingFrame.content());
        } else if (msg instanceof Http2ResetFrame) {
            Http2ResetFrame resetFrame = (Http2ResetFrame) msg;
            log.info("接收到 RST FRAME, errorCode:{}", resetFrame.errorCode());
        } else {
            log.error("unknown frame");
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("------------------exception ", cause);
    }
}
